package njuflink2;

import njuflink.GlobalConfig;
import njuflink.MySQLSink2;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.common.serialization.StringSerializer;

import java.sql.*;
import java.util.HashMap;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class KafkaFlinkMySQL2 {
    public static void main(String[] args) throws Exception{
        //获取flink运行环境
        StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
        //指定topic名称
        String topic = njuflink2.GlobalConfig.TOPIC2;
        //Kafka配置参数
        Properties properties=new Properties();
        properties.setProperty("bootstrap.servers",GlobalConfig.BOOTSRAP_SEVERS);
        properties.setProperty("group.id",topic);
        //指定key value的序列化方式
        properties.put("key.serializer", StringSerializer.class.getName());
        properties.put("value.serializer", StringSerializer.class.getName());

        //Kafka消费者
        FlinkKafkaConsumer<String> myConsumer=new FlinkKafkaConsumer<String>(topic,new SimpleStringSchema(),properties);
        DataStream<String>stream=env.addSource(myConsumer);
        System.out.println("connect kafka");
        System.out.println(myConsumer);
        //定义输出标签
        OutputTag<Tuple2<String, Long>> languageTag = new OutputTag<Tuple2<String, Long>>("language") {};
        OutputTag<Tuple2<String, Long>> licenseTag = new OutputTag<Tuple2<String, Long>>("license") {};

        SingleOutputStreamOperator<Tuple2<String, Long>> processedStream = stream.process(new ProcessFunction<String, Tuple2<String, Long>>() {
            @Override
            public void processElement(String s, Context context, Collector<Tuple2<String, Long>> out) throws Exception {
                String[] spt = s.split(",");
                //System.out.println(s+";"+spt[0]+";"+spt[1]);
                //注意String是引用，不可以直接=spt[],
                //这里比较奇怪的是，无法进行null判断，即使输出为null，可能是因为流导致的？
                //所以在生产数据的地方，进行null判断，将null的变为unknown，然后再判断是否为unknown
                String lang="",lic="";
                lang+=spt[0];
                lic+=spt[1];
//                if(lang.equals("Uknown Language")){
//                    System.out.println("lang in null");
//                    lang="Unknown Language";
//                    context.output(languageTag, Tuple2.of("Unknown Language", 1L));
//                }
////                else{
////                    System.out.println("lang not in null");
////                    context.output(languageTag, Tuple2.of(lang, 1L));
////                }
//                if(lic==null){
//                    System.out.println("lic in null");
//                    lic="Unknown License";
//                    context.output(licenseTag, Tuple2.of("Unknown License", 1L));
//                }
////                else{
////                    System.out.println("lic not in null");
////                    context.output(licenseTag, Tuple2.of(lic, 1L));
////                }
                if(!lang.equals("Unknown Language")){
                    context.output(languageTag, Tuple2.of(lang, 1L));
                }
                if(!lic.equals("Unknown License")){
                    context.output(licenseTag, Tuple2.of(lic, 1L));
                }



//                if (spt[0] != null && !spt[0].equals("")) {
//                    context.output(languageTag, Tuple2.of(spt[0], 1L));
//                }
//                if (spt[1] != null && !spt[1].equals("")) {
//                    context.output(licenseTag, Tuple2.of(spt[1], 1L));
//                }

            }
        });
//        processedStream.getSideOutput(languageTag).print("language");
//        processedStream.getSideOutput(licenseTag).print("license");
        DataStream<Tuple2<String, Long>> languagesideOutput = processedStream.getSideOutput(languageTag);
        DataStream<Tuple2<String, Long>> licensesideOutput = processedStream.getSideOutput(licenseTag);
        SingleOutputStreamOperator<Tuple2<String, Long>> languagesum = languagesideOutput.keyBy(data -> data.f0).sum(1);
        SingleOutputStreamOperator<Tuple2<String, Long>> licensesum = licensesideOutput.keyBy(data -> data.f0).sum(1);
//        languagesum.print();
//        licensesum.print();

        init_tb1();
        init_tb2();
        languagesum.addSink(new MySQLSink2_1());
        licensesum.addSink(new MySQLSink2_2());


//        KeyedStream<Tuple2<String,Long>,String> langnewCountsStream=langnewsCounts1.keyBy(data->data.f0);
//        KeyedStream<Tuple2<String, Long>,String > licnewCountsStream=licnewsCounts1.keyBy(data->data.f0);
//        SingleOutputStreamOperator<Tuple2<String, Long>> languageSum = langnewCountsStream.sum(1);
//        SingleOutputStreamOperator<Tuple2<String, Long>> licenseSum = licnewCountsStream.sum(1);
//        System.out.println("languageSum:");
//        System.out.println(languageSum);
//        languageSum.print();
//        System.out.println("licenseSum:");
//        System.out.println(licenseSum);
//        licenseSum.print();
//        languageSum.addSink(new MySQLSink2_1());
//        licenseSum.addSink(new MySQLSink2_2());

//        //自定义MySQLSink
//        periodCounts.addSink(new MySQLSink2());

        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                3, // 尝试重启的次数
                Time.of(10, TimeUnit.SECONDS) // 间隔
        ));
        //执行Flink程序
        env.execute();

    }

    public static void init_tb1() throws ClassNotFoundException, SQLException {
        Connection connection;
        PreparedStatement preparedStatement;
        String tb= njuflink2.GlobalConfig.TABLE2_1;
        //        //加载JDBC驱动
        Class.forName(GlobalConfig.DRIVER_CLASS);
        //获取数据库链接
        connection= DriverManager.getConnection(GlobalConfig.DB_URL,GlobalConfig.USER_NAME,GlobalConfig.PASSWORD);
        // 关闭自定提交
        connection.setAutoCommit(false);

//        String createsql="CREATE TABLE IF NOT EXISTS "+tb+"(language VARCHAR(50), count INT(11));";
//        System.out.println(createsql);
//        preparedStatement=connection.prepareStatement(createsql);
//        preparedStatement.executeUpdate();

        preparedStatement=connection.prepareStatement("SHOW TABLES LIKE \""+tb+"\"");
        ResultSet rs=preparedStatement.executeQuery();
//        System.out.println(rs.getFetchSize());
//        System.out.println(rs==null);
        if(rs!=null){
            System.out.println(tb+"表已存在");
            String droptbsql = "DROP TABLE "+tb;
            preparedStatement.execute(droptbsql);
            System.out.println(tb+"表已删除");
        }
        preparedStatement=connection.prepareStatement("CREATE TABLE "+tb+"(language VARCHAR(50), count INT(11)) " );
        preparedStatement.executeUpdate();
        System.out.println(tb+"表创建成功");
        connection.commit();
        if(preparedStatement!=null){
            preparedStatement.close();
        }
        if(connection!=null){
            connection.close();
        }
    }
    public static void init_tb2() throws ClassNotFoundException, SQLException {
        Connection connection;
        PreparedStatement preparedStatement;
        String tb= njuflink2.GlobalConfig.TABLE2_2;
        //        //加载JDBC驱动
        Class.forName(GlobalConfig.DRIVER_CLASS);
        //获取数据库链接
        connection= DriverManager.getConnection(GlobalConfig.DB_URL,GlobalConfig.USER_NAME,GlobalConfig.PASSWORD);
        // 关闭自定提交
        connection.setAutoCommit(false);

//        String createsql="CREATE TABLE IF NOT EXISTS "+tb+"(license VARCHAR(50), count INT(11));";
//        System.out.println(createsql);
//        preparedStatement=connection.prepareStatement(createsql);
//        preparedStatement.executeUpdate();

        preparedStatement=connection.prepareStatement("SHOW TABLES LIKE \""+tb+"\"");
        ResultSet rs=preparedStatement.executeQuery();
        if(rs!=null){
            System.out.println(tb+"表已存在");
            System.out.println(rs.getFetchSize());
            String droptbsql = "DROP TABLE "+tb;
            preparedStatement.executeUpdate(droptbsql);
            System.out.println(tb+"表已删除");
        }
        preparedStatement=connection.prepareStatement("CREATE TABLE "+tb+"(license VARCHAR(50), count INT(11)) " );
        preparedStatement.executeUpdate();
        System.out.println(tb+"表创建成功");
        connection.commit();
        if(preparedStatement!=null){
            preparedStatement.close();
        }
        if(connection!=null){
            connection.close();
        }
    }

}
